package com.hl.rpc

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable
import scala.concurrent.duration._

/**
  * Created by yuanhailong on 2021/10/8.
  */
class Master(val masterHost:String,val masterPort:Int) extends Actor{


  private val workerHashMap: mutable.Map[String, WorkerInfo] = new mutable.HashMap[String,WorkerInfo]()

  private val workerInfoSet = new mutable.HashSet[WorkerInfo]()



  override def preStart(): Unit = {
    println("preStart invoked")

    //导入隐式转换
    import context.dispatcher
    //启动定时器，定时删除死掉的Worker（超过阈值时间没有接收到心跳消息）
    context.system.scheduler.schedule(0 millis,Constants.CHECK_INTERVAL_TIMEOUT millis,self,CheckTimeoutHeartBeat)

  }




  //用于接受消息
  override def receive: Receive = {
    case RegisterWorker(workderId,memory,cores) =>{

      //如果没有注册过，就进行注册
      if(!workerHashMap.contains(workderId)){
        val workInfo = new WorkerInfo(workderId,memory,cores)

        //将Worker信息保存在内存中
        workerHashMap(workderId)=workInfo
        //将Worker信息保存在内存中，放在一个set集合中，后续好根据资源情况排序，从而Master如果有任务从有更多资源的Worker分配任务
        workerInfoSet += workInfo

        //发送Worker消息，告诉他Master已经收到消息
        sender ! RegisteredWorker(s"akka.tcp://MasterSystem@${masterHost}:${masterPort}/user/Master")
      }
    }

    //检测心跳信息，并更新最新心跳时间
    case HeartBeat(workerId)=>{
      if(workerHashMap.contains(workerId)){
        val workerInfo = workerHashMap(workerId)

        //更新最近心跳时间
        val lastTime = System.currentTimeMillis()
        workerInfo.lastHeartBeatTime=lastTime;
      }
    }

    //超时定时清理
    case CheckTimeoutHeartBeat=>{
      val currentTime=System.currentTimeMillis()

      val toRemove = workerInfoSet.filter(workInfo=>currentTime-workInfo.lastHeartBeatTime>Constants.CHECK_INTERVAL_TIMEOUT)

      //remove需要删除的数据
      for(w <- toRemove){
        workerInfoSet -= w
        workerHashMap -=w.workerId
      }

      println("workerSize:"+workerInfoSet.size)
    }



  }
}


object Master{
  def main(args: Array[String]): Unit = {
    //ActorSystem负责监控和创建下面的actor,在一个进程中只有一个ActorSystem对象（单例）
    /**
      * Actor
      *    --Actor
      *         -- actor
      *         -- actor
      *
      *    --Actor
      *         -- actor
      *         -- actor
      */


    val host=args(0)
    val port=args(1).toInt

    val configStr=
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname="${host}"
         |akka.remote.netty.tcp.port="${port}"
       """.stripMargin

    val conf:Config=ConfigFactory.parseString(configStr)

    //获取ActorSystem,他创建并监控下面的Actor
    val actorSystem: ActorSystem = ActorSystem("MasterSystem",conf)

    //创建Actor
    val master = actorSystem.actorOf(Props(new Master(host,port)),"Master")


    //给自己发送hello 消息
    //master ! "hello"

    //master.tell("hello",ActorRef.noSender)

    //等待退出
    actorSystem.awaitTermination()

  }
}
